package zju.dgm.service.dataservice;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import zju.dgm.dto.*;
import zju.dgm.service.name.NameService;
import zju.dgm.service.ServiceEntry;
import zju.dgm.util.Serializer;
import zju.dgm.util.SyncUtil;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.Deque;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.function.Function;

/**
 * @author lzy
 */
@Component
public class DataService extends ServiceEntry implements DataUtilProvider {

    private static final Logger logger = LoggerFactory.getLogger(DataService.class);

    public static final CountDownLatch finLatch = new CountDownLatch(1);

    Deque<Subscriber> subscribers = new ConcurrentLinkedDeque<>();

    BlockingQueue<DataFrame> dfHolder = new LinkedBlockingQueue<>();

    private NettyServerThread serverThread;

    private final AtomicInteger atomicInteger = new AtomicInteger(0);


    @Value("${server.port}")
    int serverPort;


    @Override
    public String getRootPath() {
        return null;
    }

    @Override
    public boolean needRootPath() {
        return false;
    }

    @Override
    protected CountDownLatch[] getPrerequisites() {
        return new CountDownLatch[]{
                NameService.finLatch
        };
    }


    @Override
    protected void registerListener() {
    }


    @Override
    protected void start() {

        serverThread = new NettyServerThread(serverPort + 1, subscribers);
        serverThread.start();
        try {
            finLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


        logger.info("数据服务加载完成");
        DataService.finLatch.countDown();

    }

    @Override
    public synchronized void sendData(ServerInfo info, DataFrame df) throws Exception {
        // 直接使用普通socket发送数据
        Socket socket;
        while (true) {
            try {
                socket = new Socket(info.getIp(), info.getPort());
                break;
            } catch (IOException e) {
                logger.warn("建立连接失败，重试，信息：{}",info);
                Thread.sleep(1000);
                e.printStackTrace();
            }
        }
        logger.info("成功建立连接");
        OutputStream os = socket.getOutputStream();
        byte[] jsonData = Serializer.getJSONString(df).getBytes(StandardCharsets.UTF_8);
        os.write(jsonData);
        os.flush();
    }

    @Override
    public DataFrame blockingGetOneDF(int millisTimeOut) throws Exception {
        return blockingGetOneDF((df) -> true, millisTimeOut);
    }


    @Override
    public DataFrame blockingGetOneDF(DFFilter filter, int millisTimeOut) throws Exception {
        Future<DataFrame> future = SyncUtil.tpe.submit(new Callable<DataFrame>() {
            @Override
            public DataFrame call() throws Exception {
                Subscriber caller = new Subscriber(atomicInteger.getAndIncrement(), filter);
                subscribers.push(caller);
                synchronized (caller.lock) {
                    caller.lock.wait();
                }
                // 只获取一次值
                subscribers.remove(caller);
                return caller.df;
            }
        });

        try {
            return future.get(millisTimeOut, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
//            e.printStackTrace();
            logger.warn("一次get调用超时！");
            return null;
        }
    }


    @PreDestroy
    void destroy() {
        //
        serverThread.bossGroup.shutdownGracefully();
        serverThread.workerGroup.shutdownGracefully();
    }


}
